package com.tping.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class SocketMultiplexingThread {

    private static int DEFAULT_PORT = 8089;

    private ServerSocketChannel server;

    private Selector bossSelector;

    private Selector workSelector;

    private Selector workSelector2;

    public void initServer(){
        try {
            server = ServerSocketChannel.open();
            server.configureBlocking(false); // 非阻塞
            server.bind(new InetSocketAddress(DEFAULT_PORT));

            bossSelector = Selector.open();
            workSelector = Selector.open();
            workSelector2 = Selector.open();

            server.register(bossSelector, SelectionKey.OP_ACCEPT);
            System.out.println(">>>>>>> 服务器启动成功.....");

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args){
        SocketMultiplexingThread socket = new SocketMultiplexingThread();
        socket.initServer();

        new ThreadNIO(socket.bossSelector, 2).start();

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new ThreadNIO(socket.workSelector).start();
        new ThreadNIO(socket.workSelector2).start();
    }

    static class ThreadNIO extends Thread{

        private Selector selector;

        private static int selectors = 0;

        private int index = 0;

        private static volatile AtomicInteger idx = new AtomicInteger(0);

        private volatile boolean bossFlag = false;

        private static BlockingQueue<SocketChannel>[] queue;

        public ThreadNIO(Selector selector, int size){
            this.selectors = size;
            this.queue = new LinkedBlockingQueue[this.selectors];
            this.selector = selector;
            this.bossFlag = true;

            for (int i = 0; i < size; i++) {
                queue[i] = new LinkedBlockingQueue<>();
            }
        }

        public ThreadNIO(Selector selector){
            this.selector = selector;
            index = idx.getAndIncrement() % selectors;
            System.out.printf("WORK: %d 启动成功....\n", index);
        }

        @Override
        public void run() {

            try {
                while (true) {
                    while ( selector.select(10) > 0 ){
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();

                        while (iterator.hasNext()){
                            SelectionKey key = iterator.next();
                            iterator.remove();

                            if( key.isAcceptable()){
                                acceptHandler(key);
                            }else if( key.isReadable() ){
                                readHandler(key);
                            }
                        }
                    }

                    if( !bossFlag && !queue[index].isEmpty()){
                        SocketChannel channel = queue[index].take();
                        ByteBuffer buffer = ByteBuffer.allocateDirect(4 * 1024);

                        channel.register(selector, SelectionKey.OP_READ, buffer);

                        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
                        System.out.printf("客户端：%s:%s 成功连接...分配到Work: %d 号\n", channel.socket().getInetAddress(), channel.socket().getPort(), index);
                        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        private void acceptHandler(SelectionKey key) {

            try {
                ServerSocketChannel server = (ServerSocketChannel) key.channel();
                SocketChannel client = server.accept();
                client.configureBlocking(false);  // 客户端非阻塞

                index = idx.getAndIncrement() % selectors;
                queue[index].add(client);
            } catch (ClosedChannelException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private void readHandler(SelectionKey key){

        }
    }

}
